package org.example.flinksql.connector;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class Datagen {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

        // create an input table
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_input (" +
                "id INT," +
                "name STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'datagen',"+
                "'rows-per-second' = '1',"+
                "'fields.id.kind' = 'sequence',"+
                "'fields.id.start' = '1',"+
                "'fields.id.end' = '19',"+
                "'fields.age.kind' = 'sequence',"+
                "'fields.age.start' = '21',"+
                "'fields.age.end'='29',"+
                "'fields.name.kind' = 'random',"+
                "'fields.name.length' = '4'," +
                "'fields.sex.kind' = 'random',"+
                "'fields.sex.length' = '2'" +
                ")");
        // get table
        Table t_input = tbEnv.from("t_input");

        /* print table struct
        root
                |-- id: INT
                |-- name: STRING
                |-- sex: STRING
                |-- age: INT
        */
        t_input.printSchema();



        // select all records from t_input
        TableResult resultA = t_input.select($("*")).execute();
        resultA.print();


        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
